package com.cmbyc.Kafka_Flink.main

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

/**
 *
 * @program: com.cmbyc.Kafka_Flink.main
 * @author: YCLW058
 * @create: 2021-06-04 16:21
 * @decsription:
 *
 * */

object Demo01 {
  def main(args: Array[String]): Unit = {
    //1 flink 环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val streamSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    val streamTableEnv = StreamTableEnvironment.create(streamEnv, streamSettings)

    //2 source
    val createInputSQL=
      """
        |CREATE TABLE Student (
        |`id` INT,
        |`name` VARCHAR(255),
        |`age` INT)
        |WITH(
        |'connector'='kafka',
        |'topic'='input_kafka',
        |'scan.startup.mode'='earliest-offset',
        |'properties.zookeeper.connect'='localhost:2181',
        |'properties.bootstrap.servers'='localhost:9092',
        |'properties.group.id'='flink'
        |""".stripMargin

    val result = streamTableEnv.executeSql(createInputSQL)

    result.print()

    streamEnv.execute()





  }

}
